package com.wzy.gmail.realtime.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import redis.clients.jedis.Jedis

import java.util
import scala.collection.mutable

/**
 * offset管理工具类，用于向redis中存储和读取offset
 *
 * 管理方案：
 *  1.后置提交偏移量->手动控制偏移量提交
 *  2.手动控制偏移量提交->sparkStream提供了手动提交方案，但是项目会对DStream进行结构转换
 *  3.手动提取偏移量维护到redis中
 *      -> 从kafka中消费到数据，先提取偏移量
 *      -> 等数据成功写出后，将偏移量存储到redis中
 *      -> 从kafka中消费数据前，先到redis中读取数据，使用读取到的偏移量到kafka中消费
 *  4.手动的将偏移量存储到redis中，每次消费数据需要存储offset进行消费，每次消费数据后，将本次消费的偏移量存储到redis中
 */
object MyOffsetUtils {

  /**
   * 在redis中存储offset
   * 问题：存的offset从哪来
   *        从消费的数据中提取出来，传入到方法中
   *        offsetRanges：Array[offsetRange]
   *      offset的结构
   *        kafka中offset维护的结构
   *          groupId + topic + partition => offset
   *        从传入进来的offset中提取关键信息
   *      在redis中怎么存
   *         类型：string list set zset hash
   *         key: groupId + topic
   *         value: partition - offset. partition -offset...
   *         写入api：hset/hmset
   *         读取api: hgetall
   *         是否过期：否
   */
  def saveOffset(topic:String,groupId:String,offsetRanges:Array[OffsetRange]):Unit={
    if(offsetRanges!=null && offsetRanges.length>0){
      val offsets = new util.HashMap[String, String]()
      for (offsetRange <- offsetRanges){
        val partition:Int = offsetRange.partition
        val endOffset:Long = offsetRange.untilOffset
        offsets.put(partition.toString,endOffset.toString)
      }
      println("提交offsets:"+offsets)
      //向redis中存储
      val jedis:Jedis = MyRedisUtils.getJedisFromPool()
      val redisKey :String = s"offsets:$topic:$groupId"
      jedis.hset(redisKey,offsets)
      jedis.close()
    }
  }

  /**
   * 从redis中读取存储的offset
   * 问题：
   *    如何让sparkStreaming通过指定的offset进行消费
   *    sparkStreaming的offset格式Map[TopicPartition,Long]
   */
  def readOffset(topic:String,groupId:String):Map[TopicPartition,Long]={
    val jedis: Jedis = MyRedisUtils.getJedisFromPool()
    val redisKey :String = s"offsets:$topic:$groupId"
    val offsets: util.Map[String, String] = jedis.hgetAll(redisKey)
    println("读取到offset:"+offsets)
    val results: mutable.Map[TopicPartition, Long] = mutable.Map[TopicPartition, Long]()
    // 将java的map转换成scala的map进行迭代
    import scala.collection.JavaConverters._
    for ((partition,offset)<- offsets.asScala){
      val tp = new TopicPartition(topic, partition.toInt)
      results.put(tp,offset.toLong)
    }
    jedis.close()
    results.toMap
  }


}
